/*
 * Copyright (c) 2006-2021, RT-Thread Development Team
 *
 * SPDX-License-Identifier: Apache-2.0
 *
 * Change Logs:
 * Date           Author       Notes
 * 2024-05-07     liuchao       the first version
 */
#include "mqtt.h"

#define LOG_TAG "mqtt"
#define LOG_LVL LOG_LVL_DBG
#include <ulog.h>

#if 0
#include <netdev_ipaddr.h>
#include <netdev.h>
#include "configuration.h"

static rt_mailbox_t s_mb;
static mqtt_client_t* s_client;

struct mqtt_msg
{
    char* topic;
    uint8_t* payload;
    uint32_t payload_len;
};

mqtt_msg_t* mqtt_msg_new(const char* topic, uint8_t** payload, uint32_t payload_len)
{
    mqtt_msg_t* ret = rt_calloc(1, sizeof(mqtt_msg_t));
    ret->topic = rt_calloc(strlen(topic) + 1, 1);
    memcpy(ret->topic, topic, strlen(topic));

    if (payload && *payload)
    {
        ret->payload = *payload;
        *payload = NULL;
        ret->payload_len = payload_len;
    }

    return ret;
}
mqtt_msg_t* mqtt_msg_new_str(const char* topic, const char* str)
{
    uint8_t* payload = rt_malloc(strlen(str));
    memcpy(payload, str, strlen(str));
    return mqtt_msg_new(topic, &payload, strlen(str));
}

#if 0
mqtt_msg_t* mqtt_msg_new_uint32(const char* topic, rt_uint32_t v)
{

}
mqtt_msg_t* mqtt_msg_new_float(const char* topic, float v);
#endif

void mqtt_msg_destroy(mqtt_msg_t* self)
{
    if (self)
    {
        rt_free(self->topic);
        rt_free(self->payload);
        rt_free(self);
    }
}

void mqtt_pub(mqtt_msg_t** msg)
{
    if (rt_mb_send(s_mb, (rt_uint32_t) *msg) == RT_EOK)
    {
    }
    else
    {
        LOG_E("send to mailbox failed");
        mqtt_msg_destroy(*msg);
    }
    *msg = NULL;
}

static void wait_for_ethernet_up()
{
    while (1)
    {
        struct netdev* n = netdev_get_by_name("e0");
        if (n && netdev_is_internet_up(n))
        {
            LOG_I("Ethernet is up");
            break;
        }
        else
        {
            LOG_D("Waiting for ethernet up");
            rt_thread_mdelay(5000);
        }
    }
}

static void prepare_mqtt_client()
{
    wait_for_ethernet_up();

    while (1)
    {
        if (mqtt_connect(s_client) == KAWAII_MQTT_SUCCESS_ERROR)
        {
            break;
        }
        else
        {
            LOG_I("Retrying mqtt connect");
            rt_thread_mdelay(5000);
        }
    }
}

void mqtt_sub(const char* topic, mqtt_qos_t qos, message_handler_t msg_handler)
{
    while (s_client == NULL)
    {
        rt_thread_mdelay(1000);
    }
    mqtt_subscribe(s_client, topic, qos, msg_handler);
}

static void mqtt_message_handler(void* client, message_data_t* msg)
{
    LOG_I("Got topic: %s, %s", msg->topic_name, msg->message->payload);
    if (strcmp(msg->topic_name, "pressureSet") == 0)
    {
        LOG_I("Got pressureSet, %s", msg->message->payload);
    }
}

void mqtt_pub_aliyun(mqtt_msg_t** msg);
void mqtt_sub_aliyun(const char* topic, mqtt_qos_t qos, message_handler_t msg_handler)
{

}

static void mqtt_entry(void *args)
{
    prepare_mqtt_client();

    mqtt_sub("/sys/k0zmlCZQpDC/864708069633447/thing/service/property/set", QOS2, mqtt_message_handler);
    //mqtt_subscribe(s_client, , qos, handler);

    mqtt_msg_t* msg = NULL;
    while (1)
    {
        if (rt_mb_recv(s_mb, (rt_uint32_t*) &msg, RT_WAITING_FOREVER) == RT_EOK)
        {
            prepare_mqtt_client();
            LOG_I("pub topic: %s", msg->topic);
            mqtt_msg_destroy(msg);
            msg = NULL;
        }
    }
}

static int mqtt_init()
{
    if (configuration_is_mqtt_host_valid())
    {
        LOG_E("mqtt host seems not right, set mqtt parameters, then reboot");
        return RT_EOK;
    }

    mqtt_log_init();
    s_client = mqtt_lease();
    mqtt_set_host(s_client, (char *) configuration_get(CONFIGURATION_MQTT_HOST));
    mqtt_set_port(s_client, (char *) configuration_get(CONFIGURATION_MQTT_PORT));
    mqtt_set_user_name(s_client, (char *) configuration_get(CONFIGURATION_MQTT_USER_NAME));
    mqtt_set_password(s_client, (char *) configuration_get(CONFIGURATION_MQTT_PASSWD));
    mqtt_set_client_id(s_client, (char *) configuration_get(CONFIGURATION_MQTT_CLIENT_ID));
    mqtt_set_clean_session(s_client, 1);

    s_mb = rt_mb_create("mqt", 64, RT_IPC_FLAG_PRIO);
    rt_thread_t t = rt_thread_create("mqt_pub", mqtt_entry, NULL, 2048, 150, 10000);
    rt_thread_startup(t);
    return RT_EOK;
}
INIT_ENV_EXPORT(mqtt_init);

#endif
